/*
 * connection.cpp
 *
 *  Created on: 2021年1月27日
 *      Author: kimfeng
 */

#include "connection.h"
#include "connection_manager.h"
#include "buf_stream.hpp"

#include <logger.h>

#include <stdlib.h>
#include <iostream>
#include <boost/bind/bind.hpp>

connection::connection(boost::asio::io_service &io_service,
		connection_manager &manager, callback_handler& handler) :
		m_sock(io_service),
		m_manager(manager),
		m_handler(handler),
		m_timer(io_service),
		m_id(0)
{
}

void connection::start(connection_id_t id)
{
	m_id = id;

	m_endpoint.append(m_sock.remote_endpoint().address().to_string());
	m_endpoint.append(":");
	m_endpoint.append(std::to_string(m_sock.remote_endpoint().port()));

	m_sock.async_read_some(boost::asio::buffer(m_buffer),
			boost::bind(&connection::handle_read, shared_from_this(),
					boost::asio::placeholders::error,
					boost::asio::placeholders::bytes_transferred));
}

void connection::stop()
{
	log_trace("connection %s will be close", m_endpoint.c_str());
	m_sock.close();
}

boost::asio::ip::tcp::socket& connection::socket()
{
	return m_sock;
}

void connection::handle_read(const boost::system::error_code &e,
		std::size_t bytes_transferred)
{
	if (e)
	{
		log_error("%s", e.message().c_str());
		log_debug("socket %s open", m_sock.is_open()?"still":"not");
		m_manager.stop(shared_from_this());
		log_debug("socket %s open after close()", m_sock.is_open()?"still":"not");
		return ;
	}

	m_timer.cancel();
	boost::tribool result;
	std::size_t bytes_parsed = 0;
	vec_data_t rsp;
	while (bytes_parsed != bytes_transferred)
	{
		boost::tie(result, boost::tuples::ignore) = m_parser.parse(
				m_pkg, m_payload,
				m_buffer.data() + bytes_parsed,
				m_buffer.data() + bytes_transferred, bytes_parsed);
		if (result)
		{
			log_trace("***get legal package***");

			CallbackFunction f = m_handler.get_callback_func(m_pkg.req_num);
			if (NULL != f)
			{
				rsp.clear();
				vec_data_t data;
				sck_pkg_t pkg;
				data.swap(m_payload);
				m_pkg.payload = data.data();
				m_pkg.conn_id = m_id;
				memcpy(&pkg, &m_pkg, sizeof(sck_pkg_t));
				memset(&m_pkg, 0, sizeof(sck_pkg_t));
				int ret = f(&pkg, rsp);
				if (0 != ret)
				{
					log_error("req_num %.4x callback function return %d", pkg.req_num, ret);
					//m_manager.stop(shared_from_this());
					//return ;
				}
				else
				{
					if (rsp.size() > 0)
					{
						//respone buffer not empty than send it back;
						send_respond(&pkg, rsp);
					}
				}
			}
			else
			{
				log_error("not registered request num %u in %s", m_pkg.req_num, m_endpoint.c_str());
			}
		}
		else if (!result)
		{
			log_error("Illegal socket package");
			m_manager.stop(shared_from_this());
			return ;
		}
		else
		{
			log_trace("bytes_transferred %llu", bytes_transferred);
			log_trace("not completely read whole package use timer to wait for other data");
			m_timer.expires_from_now(boost::posix_time::seconds(3));
			m_timer.async_wait(boost::bind(&connection::handle_rd_timeout, shared_from_this()));
		}
	}

	m_sock.async_read_some(boost::asio::buffer(m_buffer),
			boost::bind(&connection::handle_read, shared_from_this(),
					boost::asio::placeholders::error,
					boost::asio::placeholders::bytes_transferred));
}
void connection::handle_write(const boost::system::error_code &e, void *buf)
{
	if (NULL != buf)
		free(buf);

	if (e)
	{
		log_error("%s", e.message().c_str());
		m_manager.stop(shared_from_this());
	}
}

void connection::async_write(uchar *buf, std::size_t n)
{
	m_sock.async_send(boost::asio::buffer(buf, n),
			boost::bind(&connection::handle_write, shared_from_this(),
					boost::asio::placeholders::error, buf));
}

void connection::async_write(const sck_pkg_t *pkg)
{
	std::size_t n = pkg->len + PKG_HEADER_SIZE;
	uchar *buf = (uchar *)malloc(n);
	if (NULL == buf)
	{
		log_error("Not enough memory");
		return ;
	}
	uchar *p = buf;
	buf_write<ushort>(&p, pkg->magic_id);
	buf_write<ushort>(&p, pkg->ver);
	buf_write<ushort>(&p, pkg->req_num);
	buf_write<ushort>(&p, pkg->seq_num);
	buf_write<ushort>(&p, pkg->type);
	buf_write<ushort>(&p, pkg->pending);
	buf_write<uint>(&p, pkg->len);
	buf_writebytes(&p, (uchar *)pkg->payload, pkg->len);

	this->async_write(buf, n);
}

void connection::send_respond(const sck_pkg_t *pkg, const vec_data_t &rsp)
{
	std::size_t n = rsp.size() + PKG_HEADER_SIZE;
	uchar *buf = (uchar *)malloc(n);
	if (NULL != buf)
	{
		uchar *p = buf;
		buf_write<ushort>(&p, pkg->magic_id);
		buf_write<ushort>(&p, pkg->ver);
		buf_write<ushort>(&p, pkg->req_num);
		buf_write<ushort>(&p, pkg->seq_num);
		buf_write<ushort>(&p, pkg->type);
		buf_write<ushort>(&p, pkg->pending);
		buf_write<uint>(&p, rsp.size());
		buf_writebytes(&p, (uchar *)rsp.data(), rsp.size());
		async_write(buf, n);
	}
}

void connection::handle_rd_timeout()
{
	log_error("read socket package timeout at %s:%u",
			m_sock.remote_endpoint().address().to_string().c_str(),
			m_sock.remote_endpoint().port());
	stop();
}

connection_id_t connection::get_id()
{
	return m_id;
}
